
package com.yesep.learn.mqtt.client.handler.protocol;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.yesep.learn.mqtt.client.common.event.EventType;
import com.yesep.learn.mqtt.client.common.util.ChannelUtil;
import com.yesep.learn.mqtt.client.common.util.FixHeaderUtil;
import com.yesep.learn.mqtt.client.session.SessionManager;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.CharsetUtil;

/**
   * 消息编号：3
 * @Description: 发布消息
 */
public class Publish {

	private static final Logger logger = LoggerFactory.getLogger(Publish.class);

	public void process(ChannelHandlerContext ctx, MqttPublishMessage message) {
		ChannelUtil.triggerUserEvent(ctx, EventType.MESSAGE_ARRIVE, message);
		MqttQoS qos = message.fixedHeader().qosLevel();

		if (qos.equals(MqttQoS.AT_LEAST_ONCE)) {
			MqttFixedHeader fixedHeader = FixHeaderUtil.from(MqttMessageType.PUBACK);
			MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(message.variableHeader().messageId());
			MqttPubAckMessage pubAck = new MqttPubAckMessage(fixedHeader, variableHeader);
			ctx.channel().writeAndFlush(pubAck);
		} else if (qos.equals(MqttQoS.EXACTLY_ONCE)) {
			MqttFixedHeader fixedHeader = FixHeaderUtil.from(MqttMessageType.PUBREC);
			MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(message.variableHeader().messageId());
			MqttPubAckMessage pubRec = new MqttPubAckMessage(fixedHeader, variableHeader);
			ctx.channel().writeAndFlush(pubRec);
		}

		if (message.fixedHeader().isRetain()) {
			SessionManager.getSession(ChannelUtil.clientId(ctx.channel())).retainMessage(message);
		}

		ByteBuf bb = message.payload().duplicate();
		byte[] msg = new byte[bb.readableBytes()];
		bb.readBytes(msg);
		logger.info("accept message : topic-{}; content-{}", message.variableHeader().topicName(), new String(msg, CharsetUtil.UTF_8));

	}

}
